import os import time import json import subprocess import shlex # Self-contained paths relative to this script BASE_DIR = os.path.dirname(os.path.abspath(__file__)) REQUESTS_DIR = os.path.join(BASE_DIR, "comms", "requests") RESPONSES_DIR = os.path.join(BASE_DIR, "comms", "responses") SESSION_NAME = "jetski_swarm" CLI_PATH = "/google/bin/releases/jetski-devs/tools/cli" # State tracking # { target: { "last_cmd_time": timestamp, "last_output_time": timestamp, "last_output": str } } AGENT_STATE = {} HANGING_THRESHOLD = 30 # seconds of no output change while busy BUSY_THRESHOLD = 300 # seconds max execution time def get_active_agents(): try: cmd = f"tmux list-windows -t {SESSION_NAME} -F '#{{window_name}}'" result = subprocess.run(cmd, shell=True, capture_output=True, text=True, check=True) windows = result.stdout.strip().split('\n') agents = [] for w in windows: if w in ["master", "orchestrator"]: continue if w == "grid": # List panes cmd = f"tmux list-panes -t {SESSION_NAME}:grid -F '#P'" res = subprocess.run(cmd, shell=True, capture_output=True, text=True, check=True) panes = res.stdout.strip().split('\n') agents.extend([f"grid.{p}" for p in panes]) else: agents.append(w) return agents except Exception as e: print(f"Error listing agents: {e}") return [] def check_hanging(): agents = get_active_agents() now = time.time() for agent in agents: state = AGENT_STATE.setdefault(agent, { "last_cmd_time": 0, "last_output_time": now, "last_output": "" }) # Capture output for prompt detection and hanging check current_output = "" try: cap_cmd = f"tmux capture-pane -p -t {SESSION_NAME}:{agent}" result = subprocess.run(cap_cmd, shell=True, capture_output=True, text=True, check=True) current_output = result.stdout # Check for manual prompts in Orchestrator (always) if agent == "grid.0" or agent == "orchestrator": lines = current_output.split('\n') last_snippet = "\n".join(lines[-10:]) if len(lines) > 10 else current_output if "1." in last_snippet and "2." in last_snippet: if not state.get("prompted"): print(f"Detected prompt in Orchestrator.") notify_master_prompt("orchestrator", current_output) state["prompted"] = True except Exception as e: print(f"Error capturing output for {agent}: {e}") continue # Check if we recently sent a command is_busy = (now - state["last_cmd_time"]) < BUSY_THRESHOLD print(f"Debug [check_hanging]: Agent: {agent}, is_busy: {is_busy}, now: {now}, last_cmd_time: {state['last_cmd_time']}") if is_busy: print(f"Debug [check_hanging]: Agent: {agent}, current_output len: {len(current_output)}, prev_output len: {len(state['last_output'])}") if current_output != state["last_output"]: state["last_output"] = current_output state["last_output_time"] = now state["prompted"] = False # Reset prompt flag on change print(f"Debug [check_hanging]: Agent: {agent}, output changed.") else: # No change silence_duration = now - state["last_output_time"] print(f"Debug [check_hanging]: Agent: {agent}, silence_duration: {silence_duration}") if silence_duration > HANGING_THRESHOLD: print(f"Detected hanging agent: {agent}") notify_master(agent, current_output) # Reset to avoid spamming state["last_cmd_time"] = 0 def notify_master(agent, last_output): print(f"Notifying master about hanging agent {agent}...") # Extract last few lines of output lines = last_output.split('\n') last_snippet = "\n".join(lines[-10:]) if len(lines) > 10 else last_output message = f"\n[System] Warning: Agent {agent} seems to be hanging.\nLast output snippet:\n---\n{last_snippet}\n---\nWhat should we do?" try: # Send keys to master pane # We need to escape quotes in message escaped_msg = message.replace('"', '\\"') cmd = f"tmux send-keys -t {SESSION_NAME}:master \"{escaped_msg}\" C-m" subprocess.run(cmd, shell=True, check=True) except Exception as e: print(f"Failed to notify master: {e}") def notify_master_prompt(agent, last_output): print(f"Notifying master about prompt in agent {agent}...") lines = last_output.split('\n') last_snippet = "\n".join(lines[-10:]) if len(lines) > 10 else last_output message = f"\n[System] Orchestrator Brain needs input.\nPrompt snippet:\n---\n{last_snippet}\n---\nPlease reply to continue." try: escaped_msg = message.replace('"', '\\"') cmd = f"tmux send-keys -t {SESSION_NAME}:master \"{escaped_msg}\" C-m" subprocess.run(cmd, shell=True, check=True) except Exception as e: print(f"Failed to notify master: {e}") def process_request(file_path): try: with open(file_path, 'r') as f: req = json.load(f) req_id = req.get("request_id") req_type = req.get("type", "prompt") target = req.get("target_agent") or "" sender = req.get("sender") resp_file = req.get("response_file") # Enforce boundary rules if target == "master": if sender != "orchestrator": raise ValueError("Master agent can only be triggered by orchestrator agent!") if sender == "master": if target != "orchestrator": raise ValueError("Master agent can trigger orchestrator only!") if sender and (sender.startswith("grid.") or sender.startswith("agent_")): # Subagent can trigger subagent or orchestrator if target != "orchestrator" and not (target.startswith("grid.") or target.startswith("agent_")): raise ValueError("Subagents can only trigger subagent or orchestrator!") if not sender: print(f"Warning: Request {req_id} missing 'sender' field. Boundary rules not fully enforced.") # Mapping for Grid layout if target == "orchestrator": target = "grid.0" print(f"Mapped orchestrator to grid.0") if target.startswith("agent_"): try: agent_num = int(target.split("_")[1]) target = f"grid.{agent_num - 1}" print(f"Mapped {req.get('target_agent')} to {target}") except ValueError: pass print(f"Processing {req_id} (Type: {req_type}) for {target}...") output = "" status = "success" if req_type == "prompt": prompt = req.get("prompt") # Update state state = AGENT_STATE.setdefault(target, {}) state["last_cmd_time"] = time.time() # Capture output before sending to detect if it starts working cap_cmd = f"tmux capture-pane -p -t {SESSION_NAME}:{target}" try: result = subprocess.run(cap_cmd, shell=True, capture_output=True, text=True, check=True) output_before = result.stdout except Exception: output_before = "" # Send keys quoted_prompt = shlex.quote(prompt) cmd = f"tmux send-keys -t {SESSION_NAME}:{target} {quoted_prompt} C-m" subprocess.run(cmd, shell=True, check=True) # Check shortly if agent started working time.sleep(1) try: result = subprocess.run(cap_cmd, shell=True, capture_output=True, text=True, check=True) output_after = result.stdout except Exception: output_after = "" if output_before == output_after: print(f"Agent {target} seems stuck, sending extra Enter...") cmd = f"tmux send-keys -t {SESSION_NAME}:{target} C-m" subprocess.run(cmd, shell=True, check=True) time.sleep(1) # Wait a bit more # Wait for agent to process further time.sleep(3) # Capture pane for response cap_cmd = f"tmux capture-pane -p -t {SESSION_NAME}:{target}" result = subprocess.run(cap_cmd, shell=True, capture_output=True, text=True, check=True) output = result.stdout elif req_type == "control": action = req.get("action") print(f"Control Action: {action}") if action == "create": agent_dir = os.path.join(BASE_DIR, "agents", target) os.makedirs(agent_dir, exist_ok=True) conv_id = req.get("conversation_id") cli_cmd = f"{CLI_PATH} -cli=true" if conv_id: cli_cmd += f" -conversation={conv_id}" # Create tmux window cmd = f"tmux new-window -t {SESSION_NAME} -n {target} -c {agent_dir} \"{cli_cmd}\"" subprocess.run(cmd, shell=True, check=True) output = f"Created agent {target} in window and directory." if conv_id: output += f" Resumed conversation {conv_id}." elif action == "kill": cmd = f"tmux kill-window -t {SESSION_NAME}:{target}" subprocess.run(cmd, shell=True, check=True) output = f"Killed agent {target} (window)." elif action == "clear": cmd = f"tmux send-keys -t {SESSION_NAME}:{target} C-l" subprocess.run(cmd, shell=True, check=True) output = f"Sent clear screen to {target}." elif action == "resume": conv_id = req.get("conversation_id") cmd = f"tmux send-keys -t {SESSION_NAME}:{target} \"{CLI_PATH} -conversation={conv_id}\" C-m" subprocess.run(cmd, shell=True, check=True) output = f"Resumed conversation {conv_id} in {target}." else: status = "error" output = f"Unknown control action: {action}" elif req_type == "tmux": command = req.get("command") if command: if not command.startswith("tmux "): command = "tmux " + command try: subprocess.run(command, shell=True, check=True) output = f"Executed tmux command: {command}" except Exception as e: status = "error" output = f"Failed to execute tmux command: {e}" else: status = "error" output = "Missing 'command' field for tmux request type." else: status = "error" output = f"Unknown request type: {req_type}" # Write response resp_data = { "request_id": req_id, "status": status, "output": output } # Ensure target file directory exists os.makedirs(os.path.dirname(resp_file), exist_ok=True) with open(resp_file, 'w') as f: json.dump(resp_data, f, indent=2) # Delete request os.remove(file_path) print(f"Completed {req_id}") except Exception as e: print(f"Error processing {file_path}: {e}") def main(): print(f"Swarm Orchestrator started polling in {REQUESTS_DIR}...") while True: try: # Check hanging agents check_hanging() # Check requests files = os.listdir(REQUESTS_DIR) for file in files: if file.endswith(".json"): file_path = os.path.join(REQUESTS_DIR, file) process_request(file_path) except Exception as e: print(f"Polling error: {e}") time.sleep(2) if __name__ == "__main__": main()