Function send_cmd

scripts/test_swarm_debug.py:18–56

Send a debug command and return the response.

(cmd, session_id=None, timeout=10)

REPO_ROOT = Path(__file__).resolve().parent.parent

def send_cmd(cmd, session_id=None, timeout=10):
    """Send a debug command and return the response."""
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    sock.connect(SOCKET_PATH)

    req = {"type": "debug_command", "id": 1, "command": cmd}
    if session_id:
        req["session_id"] = session_id

    sock.send((json.dumps(req) + '\n').encode())

    # Read response with non-blocking to handle slow responses
    data = b''
    sock.setblocking(False)
    start = time.time()
    while time.time() - start < timeout:
        try:
            chunk = sock.recv(4096)
            if chunk:
                data += chunk
                # Check if we have a complete JSON response
                try:
                    resp = json.loads(data.decode())
                    sock.close()
                    return resp.get('ok', False), resp.get('output', '')
                except json.JSONDecodeError:
                    pass
        except BlockingIOError:
            time.sleep(0.05)

    sock.close()
    if data:
        try:
            resp = json.loads(data.decode())
            return resp.get('ok', False), resp.get('output', '')
        except json.JSONDecodeError:
            return False, f"Invalid JSON: {data.decode()[:100]}"
    raise TimeoutError("timed out")

def create_session(cwd="/tmp"):
    """Create a headless session for testing."""